#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__

#include "buffer.hpp"
#include <mutex>
#include <condition_variable>
#include <functional>
#include <memory>

using Functor = std::function<void(Buffer &)>;

enum class AsyncType
{
    ASYNC_SAFE,  // 安全状态，缓冲区满了阻塞
    ASYNC_UNSAFE // 不安全状态，无限扩容，用于性能测试
};
class AsyncLooper
{
private:
    AsyncType _looper_type;
    bool _stop;      // 工作器停止标志
    Buffer _pro_buf; // 生产缓冲区
    Buffer _con_buf; // 消费缓冲区
    mutex _mutex;
    condition_variable _cond_pro;
    condition_variable _cond_con;
    thread _thread; // 异步线程工作器对应的工作线程
private:
    Functor _callback;
    // 对消费缓冲区中的数据进行处理，处理完毕，初始化缓冲区，交换缓冲区
    void threadEntry()
    {
        while (1)
        {
            {
                unique_lock<mutex> lock(_mutex);
                if(_stop&&_pro_buf.empty())break;
                //最开始消费缓冲区没数据，所以只要生产缓冲区不为空就进行交换。
                _cond_con.wait(lock, [&]()
                               { return _stop || !_pro_buf.empty(); });
                _con_buf.swap(_pro_buf);
                _cond_pro.notify_all();
            }

            _callback(_con_buf);
            _con_buf.reset();
        }
    }

public:
    using ptr = shared_ptr<AsyncLooper>;
    AsyncLooper(const Functor &cb) : _stop(false), _thread(thread(&AsyncLooper::threadEntry, this)), _callback(cb), _looper_type(AsyncType::ASYNC_SAFE) {}
    void stop()
    {
        _stop = true;
        _cond_con.notify_all();
        _thread.join();
    }
    void push(const char *data, size_t len)
    {
        unique_lock<mutex> lock(_mutex);

        if (_looper_type == AsyncType::ASYNC_SAFE)
            _cond_pro.wait(lock, [&]()
                           { return _pro_buf.writeAbleSize() >= len; });

        _pro_buf.push(data, len);

        _cond_con.notify_one();
    }
    ~AsyncLooper()
    {
        stop();
    }
};

#endif